﻿using KafkaNet;
using KafkaNet.Model;
using KafkaNet.Protocol;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace BCGPS.DataReceive.Service
{
    public class KafakaNet
    {
        public static void Produce()
        {
            string topicName = "test";
            KafkaOptions kafkaOptions = new KafkaOptions(new Uri("http://127.0.0.1:9092"));
            Producer producer = new Producer(new BrokerRouter(kafkaOptions)) { BatchSize = 100, BatchDelayTime = TimeSpan.FromMilliseconds(2000) };
            var topic = producer.GetTopic(topicName);
            topic.Partitions = new List<KafkaNet.Protocol.Partition>(3);
            //KafkaNet.Protocol.Partition partition;
            //for ( int i=0;i<3;i++)
            //{
            //    partition = new KafkaNet.Protocol.Partition();
            //    topic.Partitions.Add(partition);
            //}

            while(true)
            {
                var message = Console.ReadLine();
                if (message == "q") break;
                if(!string.IsNullOrWhiteSpace(message))
                {
                    producer.SendMessageAsync(topicName, new[] { new Message(message) });
                  //  Console.WriteLine($"Partition: {task.Result.Partition} , Offset: {task.Result.Offset}  ");
                }
            }
        }
    }
}
